package cn.iotab.iotkit.transport.api.driver;

import cn.iotab.iotkit.common.concurrent.NamedThreadFactory;
import cn.iotab.iotkit.queue.queue.QueueConsumer;
import cn.iotab.iotkit.queue.queue.QueueProducer;
import cn.iotab.iotkit.queue.rpc.QueueRequestTemplate;
import cn.iotab.iotkit.transport.api.common.component.SingleThreadScheduler;
import cn.iotab.iotkit.transport.api.common.message.JsonQueueMsg;
import cn.iotab.iotkit.transport.api.common.message.RpcRequestMessage;
import cn.iotab.iotkit.transport.api.common.message.RpcResponseMessage;
import cn.iotab.iotkit.transport.api.common.message.ToServerMessage;
import cn.iotab.iotkit.transport.api.common.message.ToTransportMessage;
import cn.iotab.iotkit.transport.api.common.message.codec.QueueMessageFactory;
import cn.iotab.iotkit.transport.api.common.message.msg.CommonPayload;
import cn.iotab.iotkit.transport.api.common.message.msg.down.CloseSessionCmd;
import cn.iotab.iotkit.transport.api.common.message.msg.rpc.AuthBasicRequestMsg;
import cn.iotab.iotkit.transport.api.common.message.msg.rpc.AuthResponseMsg;
import cn.iotab.iotkit.transport.api.common.message.msg.rpc.AuthX509RequestMsg;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionEvent;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionInfo;
import cn.iotab.iotkit.transport.api.common.message.msg.up.SessionReportInfo;
import cn.iotab.iotkit.transport.api.common.queue.SendResultCallback;
import cn.iotab.iotkit.transport.api.common.queue.StatsCallback;
import cn.iotab.iotkit.transport.api.common.queue.TransportQueueCallback;
import cn.iotab.iotkit.transport.api.common.queue.TransportQueueFactory;
import cn.iotab.iotkit.transport.api.common.util.AsyncCallbackTemplate;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

/**
 * Created by ashvayka on 17.10.18.
 */
@Slf4j
public class DefaultTransportDriverService implements TransportDriverService {

    private final TransportDriverConfig transportDriverConf;

    private final TransportQueueFactory queueProvider;

    //定时检测session状态
    private final SingleThreadScheduler scheduler;

    //RPC template
    private QueueRequestTemplate<JsonQueueMsg<RpcRequestMessage>, JsonQueueMsg<RpcResponseMessage>> transportRpcRequestTemplate;
    // producer
    private QueueProducer<JsonQueueMsg<ToServerMessage>> transportMsgProducer;
    // consumer
    private QueueConsumer<JsonQueueMsg<ToTransportMessage>> transportNotificationsConsumer;

    //执行callback线程池
    protected ExecutorService transportCallbackExecutor;
    //单线程，pull consumer
    private ExecutorService mainConsumerExecutor;

    //session管理
    private final ConcurrentMap<String, DriverSession> sessions = new ConcurrentHashMap<>();

    private volatile boolean stopped = false;

    public DefaultTransportDriverService(TransportQueueFactory queueProvider,
                                         SingleThreadScheduler scheduler,
                                         TransportDriverConfig transportDriverConf) {
        this.queueProvider = queueProvider;
        this.scheduler = scheduler;
        this.transportDriverConf = transportDriverConf;
    }

    @PostConstruct
    public void init() {
        this.transportCallbackExecutor = Executors.newWorkStealingPool(20);
        long sessionReportTimeout = transportDriverConf.getSessionReportTimeout();
        this.scheduler.scheduleAtFixedRate(this::checkInactivityAndReportActivity, new Random().nextInt((int) sessionReportTimeout), sessionReportTimeout, TimeUnit.MILLISECONDS);

        transportRpcRequestTemplate = queueProvider.createTransportApiRequestTemplate(transportDriverConf.getRpcChannelId());
        transportMsgProducer = queueProvider.createToServerMsgProducer(transportDriverConf.getUpStreamTopic());
        transportNotificationsConsumer = queueProvider.createTransportNotificationsConsumer(transportDriverConf.getDownStreamTopic(), transportDriverConf.getConsumerGroup());

        transportRpcRequestTemplate.init();
        transportMsgProducer.init();
        transportNotificationsConsumer.subscribe();


        mainConsumerExecutor = Executors.newSingleThreadExecutor(NamedThreadFactory.forNamePrefix("transport-consumer"));
        mainConsumerExecutor.execute(() -> {
            while (!stopped) {
                try {
                    List<JsonQueueMsg<ToTransportMessage>> records = transportNotificationsConsumer.poll(transportDriverConf.getNotificationsPollDuration());
                    if (records.size() == 0) {
                        continue;
                    }
                    records.forEach(record -> {
                        try {
                            processToTransportMsg(record.getValue());
                        } catch (Throwable e) {
                            log.warn("Failed to process the notification.", e);
                        }
                    });
                    transportNotificationsConsumer.commit();
                } catch (Exception e) {
                    if (!stopped) {
                        log.warn("Failed to obtain messages from queue.", e);
                        try {
                            Thread.sleep(transportDriverConf.getNotificationsPollDuration());
                        } catch (InterruptedException e2) {
                            log.trace("Failed to wait until the server has capacity to handle new requests", e2);
                        }
                    }
                }
            }
        });
    }

    @PreDestroy
    public void destroy() {
        stopped = true;

        if (transportNotificationsConsumer != null) {
            transportNotificationsConsumer.unsubscribe();
        }
        if (transportCallbackExecutor != null) {
            transportCallbackExecutor.shutdownNow();
        }
        if (mainConsumerExecutor != null) {
            mainConsumerExecutor.shutdownNow();
        }
        if (transportRpcRequestTemplate != null) {
            transportRpcRequestTemplate.stop();
        }
    }

    @Override
    public DriverSession registerSession(SessionInfo sessionInfo, SessionMsgListener listener) {
        DriverSession driverSession = new DriverSession(sessionInfo, listener);
        sessions.putIfAbsent(sessionInfo.getSessionId(), driverSession);
        return driverSession;
    }

    @Override
    public void authenticate(AuthBasicRequestMsg msg, SendResultCallback<AuthResponseMsg> callback) {
        log.trace("Processing msg: {}", msg);
        RpcRequestMessage requestMessage = RpcRequestMessage.builder().authBasic(msg).build();
        JsonQueueMsg<RpcRequestMessage> reqQueueMsg = QueueMessageFactory.createQueueMessage(msg.getClientId(), requestMessage);
        doProcessRpc(reqQueueMsg, RpcResponseMessage::getAuthResponse, callback);
    }

    @Override
    public void authenticate(AuthX509RequestMsg msg, SendResultCallback<AuthResponseMsg> callback) {
        log.trace("Processing msg: {}", msg);
        RpcRequestMessage requestMessage = RpcRequestMessage.builder().authX509(msg).build();
        JsonQueueMsg<RpcRequestMessage> reqQueueMsg = QueueMessageFactory.createQueueMessage(msg.getClientId(), requestMessage);
        doProcessRpc(reqQueueMsg, RpcResponseMessage::getAuthResponse, callback);
    }

    /**
     * 处理RPC请求
     * @param rpcReqMessage 请求Message
     * @param resultPicker 从RpcResponse中拿出Response
     * @param callback 回调
     * @param <T> Response对象类型
     */
    private <T> void doProcessRpc(JsonQueueMsg<RpcRequestMessage> rpcReqMessage,
                              Function<RpcResponseMessage, T> resultPicker,
                              SendResultCallback<T> callback) {
        ListenableFuture<JsonQueueMsg<RpcResponseMessage>> responseFuture = transportRpcRequestTemplate.send(rpcReqMessage);
        ListenableFuture<T> response = Futures.transform(responseFuture, tmp -> resultPicker.apply(tmp.getValue()), MoreExecutors.directExecutor());
        //
        AsyncCallbackTemplate.withCallback(response, callback::onSuccess, callback::onError, transportCallbackExecutor);
    }

    @Override
    public void reportSessionInfo(SessionInfo sessionInfo, SessionReportInfo msg, SendResultCallback<Void> callback) {
        if (log.isTraceEnabled()) {
            log.trace("[{}] Processing msg: {}", sessionInfo.getSessionId(), msg);
        }
        ToServerMessage toServerMessage = ToServerMessage.builder()
                .sessionInfo(sessionInfo)
                .sessionReportInfo(msg)
                .build();
        sendToServer(sessionInfo, toServerMessage, callback);
    }

    @Override
    public void onSessionStateChange(SessionInfo sessionInfo, SessionEvent msg, SendResultCallback<Void> callback) {
        updateSessionActiveTime(sessionInfo);
        ToServerMessage toServerMessage = ToServerMessage.builder()
                .sessionInfo(sessionInfo)
                .sessionEvent(msg)
                .build();
        sendToServer(sessionInfo, toServerMessage, callback);
    }

    @Override
    public void onDeviceRequestMessage(SessionInfo sessionInfo, CommonPayload msg, SendResultCallback<Void> callback) {
        updateSessionActiveTime(sessionInfo);

        ToServerMessage toServerMessage = ToServerMessage.builder()
                .sessionInfo(sessionInfo)
                .fromDeviceRequest(msg).build();
        sendToServer(sessionInfo, toServerMessage, callback);
    }

    @Override
    public void onDeviceResponseMessage(SessionInfo sessionInfo, CommonPayload msg, SendResultCallback<Void> callback) {
        updateSessionActiveTime(sessionInfo);

        ToServerMessage toServerMessage = ToServerMessage.builder()
                .sessionInfo(sessionInfo)
                .fromDeviceResponse(msg).build();
        sendToServer(sessionInfo, toServerMessage, callback);
    }

    @Override
    public void sessionActivity(SessionInfo sessionInfo) {
        updateSessionActiveTime(sessionInfo);
    }

    private DriverSession updateSessionActiveTime(SessionInfo sessionInfo) {
        String sessionId = sessionInfo.getSessionId();
        DriverSession driverSession = sessions.get(sessionId);
        if (driverSession != null) {
            driverSession.updateLastActivityTime();
        }
        return driverSession;
    }

    private void checkInactivityAndReportActivity() {
        long expTime = System.currentTimeMillis() - transportDriverConf.getSessionInactivityTimeout();
        sessions.forEach((uuid, sessionMD) -> {
            long lastActivityTime = sessionMD.getLastActivityTime();
            SessionInfo sessionInfo = sessionMD.getSessionInfo();
            if (lastActivityTime < expTime) {
                if (log.isDebugEnabled()) {
                    log.debug("[{}] Session has expired due to last activity time: {}", sessionInfo.getSessionId(), lastActivityTime);
                }
                onSessionStateChange(sessionInfo, SessionEvent.CLOSED, null);
                sessions.remove(uuid);
                sessionMD.getListener().onRemoteSessionCloseCommand(new CloseSessionCmd());
            } else {
                if (lastActivityTime > sessionMD.getLastReportedActivityTime()) {
                    final long lastActivityTimeFinal = lastActivityTime;
                    SessionReportInfo reportInfo = SessionReportInfo.builder()
                            .subscribedMessageTopic(sessionMD.isSubscribedMessageTopic())
                            .subscribedReplyTopic(sessionMD.isSubscribedReplyTopic())
                            .lastActivityTime(lastActivityTime).build();

                    reportSessionInfo(sessionInfo, reportInfo, new SendResultCallback<Void>() {
                        @Override
                        public void onSuccess(Void msg) {
                            sessionMD.setLastReportedActivityTime(lastActivityTimeFinal);
                        }

                        @Override
                        public void onError(Throwable e) {
                            log.warn("[{}] Failed to report last activity time", uuid, e);
                        }
                    });
                }
            }
        });
    }

    @Override
    public void deregisterSession(SessionInfo sessionInfo) {
        sessions.remove(sessionInfo.getSessionId());
    }


    protected void processToTransportMsg(ToTransportMessage toTransportMessage) {
        String sessionId = toTransportMessage.getSessionId();
        if (StringUtils.isBlank(sessionId)){
            log.error("sessionId is null: {}", toTransportMessage);
        }
        DriverSession md = sessions.get(sessionId);
        if (md != null) {
            SessionMsgListener listener = md.getListener();
            transportCallbackExecutor.submit(() -> {
                if (toTransportMessage.getCloseSessionCmd() != null) {
                    listener.onRemoteSessionCloseCommand(toTransportMessage.getCloseSessionCmd());
                    return;
                }
                if (toTransportMessage.getToDeviceRequest() != null) {
                    listener.onToDeviceRequest(toTransportMessage.getToDeviceRequest());
                    return;
                }
                if (toTransportMessage.getToDeviceResponse() != null) {
                    listener.onToDeviceResponse(toTransportMessage.getToDeviceResponse());
                    return;
                }
                log.error("unknown ToTransportMessage:{}", toTransportMessage);
            });
        } else {
            log.error("session not exist: {}", sessionId);
        }
    }



    private void sendToServer(SessionInfo sessionInfo, ToServerMessage toServerMessage, SendResultCallback<Void> callback) {
        if (log.isTraceEnabled()) {
            log.trace("[{}] Pushing message to core: {}", sessionInfo.getSessionId(), toServerMessage);
        }
        TransportQueueCallback transportQueueCallback = callback != null ? new TransportQueueCallback(callback, transportCallbackExecutor) : null;
        StatsCallback wrappedCallback = new StatsCallback(transportQueueCallback);
        //
        toServerMessage.setSessionInfo(sessionInfo);
        JsonQueueMsg<ToServerMessage> msg = new JsonQueueMsg<>(sessionInfo.getSessionId(), toServerMessage);
        transportMsgProducer.send(msg, wrappedCallback);
    }



}
